package realtime.app.dwd.db;

import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import realtime.util.MyKafkaUtil;

public class Dwd_TradePayDetailSus {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        /*env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flinkCDC");
        env.setStateBackend(new HashMapStateBackend());*/

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("" +
                        "create table dwd_trade_order_detail(\n" +
                        "id String,\n" +
                        "order_id String,\n" +
                        "user_id String,\n" +
                        "course_id String,\n" +
                        "course_name String,\n" +
                        "province_id String,\n" +
                        "create_date String,\n" +
                        "create_time String,\n" +
                        "session_id String,\n" +
                        "update_time String,\n" +
                        "expire_time String,\n" +
                        "origin_amount String,\n" +
                        "coupon_reduce String,\n" +
                        "final_amount String,\n" +
                        "ts BIGINT,\n" +
                        "row_time as TO_TIMESTAMP(FROM_UNIXTIME(ts)),\n" +
                        "watermark for row_time as row_time\n" +
                        ")"  + MyKafkaUtil.getKafkaDDL("dwd_trade_order_detail","trade_pay_detail_sus"));

        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("trade_pay_detail_sus"));

        Table paymentInfoTable = tableEnv.sqlQuery("select\n" +
                "\tdata['id'] id,\n" +
                "\tdata['out_trade_no'] out_trade_no,\n" +
                "\tdata['order_id'] order_id,\n" +
                "\tdata['alipay_trade_no'] alipay_trade_no,\n" +
                "\tdata['total_amount'] total_amount,\n" +
                "\tdata['trade_body'] trade_body,\n" +
                "\tdata['payment_type'] payment_type,\n" +
                "\tdata['payment_status'] payment_status,\n" +
                "\tdata['create_time'] create_time,\n" +
                "\tdata['update_time'] update_time,\n" +
                "\tdata['callback_content'] callback_content,\n" +
                "\tdata['callback_time'] callback_time,\n" +
                "\tts\n" +
                "from topic_db\n" +
                "where `table` = 'payment_info'\n" +
                "and type = 'update'\n" +
                "and data['payment_status'] = '1602'");
        tableEnv.createTemporaryView("pay_info",paymentInfoTable);
        tableEnv.sqlQuery("select * from pay_info").execute().print();

        Table resultTable = tableEnv.sqlQuery("" +
                "select\n" +
                "\tod.id order_detail_id,\n" +
                "\tod.order_id,\n" +
                "\tod.user_id,\n" +
                "\tod.course_id,\n" +
                "\tod.course_name,\n" +
                "\tod.province_id,\n" +
                "\tod.create_date,\n" +
                "\tod.create_time,\n" +
                "\tod.session_id,\n" +
                "\tod.expire_time,\n" +
                "\tod.origin_amount,\n" +
                "\tod.coupon_reduce,\n" +
                "\tod.final_amount,\n" +
                "\tpi.out_trade_no,\n" +
                "\tpi.alipay_trade_no,\n" +
                "\tpi.total_amount,\n" +
                "\tpi.trade_body,\n" +
                "\tpi.payment_type,\n" +
                "\tpi.payment_status,\n" +
                "\tpi.callback_content,\n" +
                "\tpi.callback_time,\n" +
                "\tpi.ts\n" +
                "from pay_info pi\n" +
                "join dwd_trade_order_detail od\n" +
                "on pi.order_id = od.order_id");

        tableEnv.createTemporaryView("result_table",resultTable);

        tableEnv.executeSql("create table dwd_trade_pay_detail_sus(\n" +
                "\torder_detail_id string,\n" +
                "\torder_id string,\n" +
                "\tuser_id string,\n" +
                "\tcourse_id string,\n" +
                "\tcourse_name string,\n" +
                "\tprovince_id string,\n" +
                "\tcreate_date string,\n" +
                "\tcreate_time string,\n" +
                "\tsession_id string,\n" +
                "\texpire_time string,\n" +
                "\torigin_amount string,\n" +
                "\tcoupon_reduce string,\n" +
                "\tfinal_amount string,\n" +
                "\tout_trade_no string,\n" +
                "\talipay_trade_no string,\n" +
                "\ttotal_amount string,\n" +
                "\ttrade_body string,\n" +
                "\tpayment_type string,\n" +
                "\tpayment_status string,\n" +
                "\tcreate_time string,\n" +
                "\tcallback_content string,\n" +
                "\tcallback_time string,\n" +
                "\tts bigint,\n" +
                "primary key(order_detail_id) not enforced " +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_pay_detail_sus"));

        tableEnv.executeSql("insert into dwd_trade_pay_detail_sus select * from result_table");
    }
}
